package com.xj.driver.mqtt.subscriber.queue;

import com.xj.driver.mqtt.subscriber.monitor.MessageBacklogMonitor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.Message;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 消息队列管理器
 */
@Slf4j
public class MessageQueueManager {

    private static final int QUEUE_CAPACITY = 10000;

    private final BlockingQueue<Message<?>> messageBuffer;
    private final ExecutorService processorExecutor;
    private final MessageProcessor messageProcessor;
    private final MessageBacklogMonitor monitor;

    public MessageQueueManager(MessageProcessor messageProcessor, MessageBacklogMonitor monitor) {
        this.messageBuffer = new LinkedBlockingQueue<>(QUEUE_CAPACITY);
        this.processorExecutor = Executors.newFixedThreadPool(1,
                r -> new Thread(r, "MqttMessageProcessor"));
        this.messageProcessor = messageProcessor;
        this.monitor = monitor;
    }


    /**
     * 添加消息到队列
     */
    public void addMessage(Message<?> message) {
        try {
            // 检查队列容量
            if (messageBuffer.remainingCapacity() == 0) {
                log.error("🚨 消息队列已满，无法接收新消息");
                monitor.recordMessageDiscarded();
                return;
            }
            messageBuffer.put(message);
            monitor.recordMessageReceived();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("消息入队被中断", e);
            monitor.recordMessageDiscarded();
        }
    }

    /**
     * 启动消息处理
     */
    public void startProcessing() {
        processorExecutor.submit(this::processMessageQueue);
    }

    /**
     * 停止消息处理
     */
    public void stopProcessing() {
        processorExecutor.shutdownNow();
    }

    /**
     * 处理消息队列
     */
    private void processMessageQueue() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                Message<?> message = messageBuffer.take();
                try {
                    messageProcessor.process(message);
                    monitor.recordMessageProcessed();
                } catch (Exception e) {
                    log.error("处理消息内容时发生异常: {}", message, e);
                    monitor.recordMessageDiscarded();
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            } catch (Exception e) {
                log.error("消息队列处理异常", e);
            }
        }
    }

    /**
     * 消息处理接口
     */
    public interface MessageProcessor {
        void process(Message<?> message);
    }
}
